package org.jetlinks.pro.things.impl.connector;

import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
import org.hswebframework.web.bean.FastBeanCopier;
import org.hswebframework.web.id.IDGenerator;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.pro.streaming.Streaming;
import org.jetlinks.pro.things.ProviderProperties;
import org.jetlinks.pro.things.ThingsConnector;
import org.jetlinks.pro.things.ThingsConnectorProvider;
import org.jetlinks.pro.things.ThingsManagerContext;
import org.jetlinks.pro.things.impl.RecursiveCheckThingsConnector;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import reactor.core.publisher.Mono;

import java.util.Map;

@AllArgsConstructor
@Component
public class ComputeThingsConnectorProvider implements ThingsConnectorProvider {

    private final ThingsManagerContext context;

    @Override
    public String getId() {
        return "compute";
    }

    @Override
    public String getName() {
        return "规则计算";
    }

    @Override
    public Mono<ThingsConnector> createConnector(Map<String, Object> configuration) {
        ComputeConnectorProperties properties = FastBeanCopier.copy(configuration, new ComputeConnectorProperties());
        properties.validate();

        //创建消息计算器
        Streaming<ThingMessage, ComputeThingsConnector.ComputeResult, ComputeThingsConnector.ComputeResult> streaming =
            properties.rule.create(ComputeThingsConnector.ComputeResult::of);

        //将MD5随机数作为当前规则计算连接器的跟踪标识。 用于检查连接器是否存在递归
        String trace = IDGenerator.MD5.generate();
        return context
            .createConnector(properties.source.getProvider(), properties.source.getConfiguration())
            .map(source -> new ComputeThingsConnector(RecursiveCheckThingsConnector.of(trace,source), streaming));
    }

    @Getter
    @Setter
    public static class ComputeConnectorProperties {

        /**
         * 源连接器
         */
        private ProviderProperties source;

        /**
         * 计算规则
         */
        private ComputeRule rule;

        public void validate() {
            Assert.notNull(source, "source can not be null");
            source.validate();
            Assert.notNull(rule, "rule can not be null");
            rule.validate();
        }
    }


}
